using Gasgoo.ShareLibrary.Framework.Global;
using Gasgoo.ShareLibrary.Framework.iQueueProvider;
using Gasgoo.ShareLibrary.Framework.Options;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Orleans;
using Orleans.Configuration;
using Orleans.Providers;
using Orleans.Providers.Streams.Common;
using Orleans.Runtime;
using Orleans.Streams;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;

namespace Gasgoo.ShareLibrary.Framework.iQueueProvider
{

    public class AdapterFactory : IAdapterFactory
    {
        public IQueueAdapterFactory Create(IServiceProvider services, string name)
        {
            var cachePurgeOptions = services.GetOptionsByName<StreamCacheEvictionOptions>(name);
            var statisticOptions = services.GetOptionsByName<StreamStatisticOptions>(name);
            var queueMapperOptions = services.GetOptionsByName<HashRingStreamQueueMapperOptions>(name);
            var factory = ActivatorUtilities.CreateInstance<Adapter>(services, name, cachePurgeOptions, statisticOptions, queueMapperOptions);
            factory.Init();
            return factory;
        }
    }

    /// <summary>
    /// Adapter factory for in memory stream provider.
    /// This factory acts as the adapter and the adapter factory.  The events are stored in an in-memory grain that 
    /// behaves as an event queue, this provider adapter is primarily used for testing
    /// </summary>
    public class Adapter : IQueueAdapterFactory, IQueueAdapter, IQueueAdapterCache
    {
        private readonly StreamCacheEvictionOptions cacheOptions;
        private readonly StreamStatisticOptions statisticOptions;
        private readonly HashRingStreamQueueMapperOptions queueMapperOptions;
        private readonly IGrainFactory grainFactory;
        private readonly ITelemetryProducer telemetryProducer;
        private readonly ILoggerFactory loggerFactory;
        private readonly ILogger logger;
        private readonly DefaultMemoryMessageBodySerializer serializer;
        private IStreamQueueMapper streamQueueMapper;
        IQueueStorageProvider queueStorageProvider { get; }
        private ConcurrentDictionary<QueueId, IQueueStorageProvider> queueStorageProviders;
        private IObjectPool<FixedSizeBuffer> bufferPool;
        private BlockPoolMonitorDimensions blockPoolMonitorDimensions;
        private IStreamFailureHandler streamFailureHandler;
        private TimePurgePredicate purgePredicate;
        public string Name { get; }
        public bool IsRewindable => true;
        public StreamProviderDirection Direction => StreamProviderDirection.ReadWrite;
        public IServiceProvider serviceProvider { get; }
        AdoNetOptions options { get; }
        protected Func<string, Task<IStreamFailureHandler>> StreamFailureHandlerFactory { get; set; }
        protected Func<CacheMonitorDimensions, ITelemetryProducer, ICacheMonitor> CacheMonitorFactory;
        protected Func<BlockPoolMonitorDimensions, ITelemetryProducer, IBlockPoolMonitor> BlockPoolMonitorFactory;
        protected Func<ReceiverMonitorDimensions, ITelemetryProducer, IQueueAdapterReceiverMonitor> ReceiverMonitorFactory;

        public Adapter(string providerName, StreamCacheEvictionOptions cacheOptions, StreamStatisticOptions statisticOptions, HashRingStreamQueueMapperOptions queueMapperOptions,
            IServiceProvider serviceProvider, IGrainFactory grainFactory, ITelemetryProducer telemetryProducer, ILoggerFactory loggerFactory)
        {
            this.Name = providerName;
            this.queueMapperOptions = queueMapperOptions ?? throw new ArgumentNullException(nameof(queueMapperOptions));
            this.cacheOptions = cacheOptions ?? throw new ArgumentNullException(nameof(cacheOptions));
            this.statisticOptions = statisticOptions ?? throw new ArgumentException(nameof(statisticOptions));
            this.grainFactory = grainFactory ?? throw new ArgumentNullException(nameof(grainFactory));
            this.telemetryProducer = telemetryProducer ?? throw new ArgumentNullException(nameof(telemetryProducer));
            this.loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory));
            this.logger = loggerFactory.CreateLogger<ILogger<AdapterFactory>>();
            this.serializer = MessageBodySerializerFactory<DefaultMemoryMessageBodySerializer>.GetOrCreateSerializer(serviceProvider);
            this.serviceProvider = serviceProvider;
            this.options = this.serviceProvider.GetRequiredService<AdoNetOptions>();

            this.queueStorageProvider = this.serviceProvider.GetRequiredServiceByName<IQueueStorageProvider>(providerName);
            this.queueStorageProviders = new ConcurrentDictionary<QueueId, IQueueStorageProvider>();

            var collection = ServiceManage.GetServiceCollection(Constants.ITOOL_SERVICE);
            collection.AddSingleton<DefaultMemoryMessageBodySerializer>(this.serializer);
            collection.AddSingleton<IGrainFactory>(grainFactory);
        }

        /// <summary>
        /// Factory initialization.
        /// </summary>
        public void Init()
        {
            if (CacheMonitorFactory == null)
                this.CacheMonitorFactory = (dimensions, telemetryProducer) => new DefaultCacheMonitor(dimensions, telemetryProducer);
            if (this.BlockPoolMonitorFactory == null)
                this.BlockPoolMonitorFactory = (dimensions, telemetryProducer) => new DefaultBlockPoolMonitor(dimensions, telemetryProducer);
            if (this.ReceiverMonitorFactory == null)
                this.ReceiverMonitorFactory = (dimensions, telemetryProducer) => new DefaultQueueAdapterReceiverMonitor(dimensions, telemetryProducer);
            this.purgePredicate = new TimePurgePredicate(this.cacheOptions.DataMinTimeInCache, this.cacheOptions.DataMaxAgeInCache);
            this.streamQueueMapper = new HashRingBasedStreamQueueMapper(this.queueMapperOptions, this.Name);
        }

        private void CreateBufferPoolIfNotCreatedYet()
        {
            if (this.bufferPool == null)
            {
                // 1 meg block size pool
                this.blockPoolMonitorDimensions = new BlockPoolMonitorDimensions($"BlockPool-{Guid.NewGuid()}");
                var oneMb = 1 << 20;
                var objectPoolMonitor = new ObjectPoolMonitorBridge(this.BlockPoolMonitorFactory(blockPoolMonitorDimensions, this.telemetryProducer), oneMb);
                this.bufferPool = new ObjectPool<FixedSizeBuffer>(() => new FixedSizeBuffer(oneMb), objectPoolMonitor, this.statisticOptions.StatisticMonitorWriteInterval);
            }
        }

        /// <summary>
        /// Create queue adapter.
        /// </summary>
        /// <returns></returns>
        public Task<IQueueAdapter> CreateAdapter()
        {
            return Task.FromResult<IQueueAdapter>(this);
        }

        /// <summary>
        /// Create queue message cache adapter
        /// </summary>
        /// <returns></returns>
        public IQueueAdapterCache GetQueueAdapterCache()
        {
            return this;
        }

        /// <summary>
        /// Create queue mapper
        /// </summary>
        /// <returns></returns>
        public IStreamQueueMapper GetStreamQueueMapper()
        {
            return streamQueueMapper;
        }

        /// <summary>
        /// Creates a queue receiver for the specified queueId
        /// </summary>
        /// <param name="queueId"></param>
        /// <returns></returns>
        public IQueueAdapterReceiver CreateReceiver(QueueId queueId)
        {
            var dimensions = new ReceiverMonitorDimensions(queueId.ToString());
            var receiverLogger = this.loggerFactory.CreateLogger($"{typeof(AdapterReceiver<DefaultMemoryMessageBodySerializer>).FullName}.{this.Name}.{queueId}");
            var receiverMonitor = this.ReceiverMonitorFactory(dimensions, this.telemetryProducer);
            IQueueAdapterReceiver receiver = new AdapterReceiver<DefaultMemoryMessageBodySerializer>(Name, grainFactory, queueId.ToString(), receiverLogger, this.serializer, this.serviceProvider, receiverMonitor);
            return receiver;
        }

        /// <summary>
        /// Create a cache for a given queue id
        /// </summary>
        /// <param name="queueId"></param>
        public IQueueCache CreateQueueCache(QueueId queueId)
        {
            CreateBufferPoolIfNotCreatedYet();
            var logger = this.loggerFactory.CreateLogger($"{typeof(PooledCache<DefaultMemoryMessageBodySerializer>).FullName}.{this.Name}.{queueId}");
            var monitor = this.CacheMonitorFactory(new CacheMonitorDimensions(queueId.ToString(), this.blockPoolMonitorDimensions.BlockPoolId), this.telemetryProducer);
            return new PooledCache<DefaultMemoryMessageBodySerializer>(bufferPool, purgePredicate, logger, this.serializer, monitor, this.statisticOptions.StatisticMonitorWriteInterval);
        }

        /// <summary>
        /// Acquire delivery failure handler for a queue
        /// </summary>
        /// <param name="queueId"></param>
        /// <returns></returns>
        public Task<IStreamFailureHandler> GetDeliveryFailureHandler(QueueId queueId)
        {
            Console.WriteLine("GetDeliveryFailureHandler queueId is {0}", queueId);


            return Task.FromResult<IStreamFailureHandler>(new StreamDeliveryFailureHandler());
            return Task.FromResult(streamFailureHandler ?? (streamFailureHandler = new NoOpStreamDeliveryFailureHandler()));
        }
        private IQueueStorageProvider GetQueueStorageProvider(QueueId queueId)
        {
            return queueStorageProviders.GetOrAdd(queueId, id => this.queueStorageProvider.CreateInstance(this.serviceProvider, queueId.ToString()));
        }

        public async Task QueueMessageBatchAsync<T>(Guid streamGuid, string streamNamespace, IEnumerable<T> events, StreamSequenceToken token, Dictionary<string, object> requestContext)
        {
            try
            {
                var queueId = streamQueueMapper.GetQueueForStream(streamGuid, streamNamespace);
                var bodyBytes = serializer.Serialize(new MemoryMessageBody(events.Cast<object>(), requestContext));
                var messageData = MessageDataOptions.Create(new StreamIdentity(streamGuid, streamNamespace), bodyBytes);
                IQueueStorageProvider queueStorage = GetQueueStorageProvider(queueId);
                await queueStorage.EnqueueAsync(messageData);
            }
            catch (Exception)
            {
                throw;
            }
        }
    }
}
